package david.java.flink_sql.streamApiIntegration;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import java.time.LocalDateTime;

/**
 * @Description: The following Java example illustrates a Flink program that is not only unified on an API level but also in the resulting changelog stream.
 * The example joins two tables in SQL (UserTable and OrderTable) using an interval join based on the time attributes in both tables (ts).
 * It uses DataStream API to implement a custom operator that deduplicates the user name using a KeyedProcessFunction and value state.
 * @Author: ZhaoDawei
 * @Date: Create in 10:43 上午 2022/3/25
 */
public class S2_ChangeLogUnification {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // create a user steam
        DataStream<Row> userStream = env.fromElements(
                Row.of(LocalDateTime.parse("2021-08-21T13:00:00"), 1, "Alice"),
                Row.of(LocalDateTime.parse("2021-08-21T13:05:00"), 2, "Bob"),
                Row.of(LocalDateTime.parse("2021-08-21T14:00:00"), 1, "Alice"),
                Row.of(LocalDateTime.parse("2021-08-21T14:10:00"), 2, "Bob"))
                .returns(
                        Types.ROW_NAMED(
                                new String[]{"ts", "uid", "name"},
                                Types.LOCAL_DATE_TIME, Types.INT, Types.STRING));
        // create an order stream
        DataStream<Row> orderStream = env
                .fromElements(
                        Row.of(LocalDateTime.parse("2021-08-21T13:02:00"), 1, 122),
                        Row.of(LocalDateTime.parse("2021-08-21T13:07:00"), 2, 239),
                        Row.of(LocalDateTime.parse("2021-08-21T14:02:00"), 1, 888),
                        Row.of(LocalDateTime.parse("2021-08-21T14:11:00"), 2, 999))
                .returns(
                        Types.ROW_NAMED(
                                new String[]{"ts", "uid", "amount"},
                                Types.LOCAL_DATE_TIME, Types.INT, Types.INT));

        tableEnv.createTemporaryView(
                "OrderTable",
                orderStream,
                Schema.newBuilder()
                        .column("ts", DataTypes.TIMESTAMP(3))
                        .column("uid", DataTypes.INT())
                        .column("amount", DataTypes.INT())
                        .watermark("ts", "ts - INTERVAL '1' SECOND")
                        .build());

        // create corresponding tables
        tableEnv.createTemporaryView(
                "UserTable",
                userStream,
                Schema.newBuilder()
                        .column("ts", DataTypes.TIMESTAMP(3))
                        .column("uid", DataTypes.INT())
                        .column("name", DataTypes.STRING())
                        .watermark("ts", "ts - INTERVAL '1' SECOND")
                        .build());

        Table joinedTable = tableEnv.sqlQuery(
                "SELECT U.name, O.amount " +
                        "FROM UserTable U, OrderTable O " +
                        "WHERE U.uid = O.uid AND O.ts BETWEEN U.ts AND U.ts + INTERVAL '5' MINUTES"
        );

        DataStream<Row> joinedStream = tableEnv.toDataStream(joinedTable);
        // 12> +I[Bob, 239]
        // 12> +I[Bob, 999]
        // 7> +I[Alice, 122]
        // 7> +I[Alice, 888]
        joinedStream.print();


        joinedStream
                .keyBy(r -> r.<String>getFieldAs("name"))
                .process(new KeyedProcessFunction<String, Row, String>() {
                    // 只能用在keyed算子里面
                    ValueState<String> seen;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        seen = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("seen", String.class));
                    }

                    @Override
                    public void processElement(Row row, Context ctx, Collector<String> out) throws Exception {
                        String name = row.getFieldAs("name");
                        if (seen.value() == null) {
                            seen.update(name);
                            out.collect(name);
                        }
                    }
                })
                // 12> Bob
                // 9> Alice
                .print();





        env.execute();


    }
}
